
/***************************************************************************
 *   jell - library                                                        *
 *                                                                         *
 *   Copyright (C) 2008 by kuerant                                         *
 *                         <kuerant@gmail.com>                             *
 *                                                                         *
 *   This library is free software; you can redistribute it and/or modify  *
 *   it under the terms of the GNU Library General Public License as       *
 *   published by the Free Software Foundation; either version 2 of the    *
 *   License, or (at your option) any later version.                       *
 *                                                                         *
 *   This library is distributed in the hope that it will be usefull, but  *
 *   WITHOUT ANY WARRANTY; without even the implied warranty of            *
 *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU      *
 *   Library General Public License for more details.                      *
 *                                                                         *
 *   You should have received a copy of the GNU Library General Public     *
 *   License along with this library; if not, write to the Free Software   *
 *   Foundation, Inc., 59 Temple Place - Suite 330, Boston,                *
 *   MA 02111-1307, USA                                                    *
 ***************************************************************************/


#include "jell/jelly.hpp"


namespace   jell
{

///////////////////////////////////////////////////////////////////////

static const char*  TCP_NAME  = "tcp";
static const char*  UDP_NAME  = "udp";
static const char*  UNIX_NAME = "unix";

static const std::string    UDP_ANY_URI( "udp://any" );

///////////////////////////////////////////////////////////////////////

static  bool      packet_verifyHead(void* packet_head);
static  unsigned  packet_getBodySize(void* packet_head);
static  void      packet_setBodySize(void* packet_head, unsigned data_size);

static  std::string dump_hex(void* data, size_t size);

///////////////////////////////////////////////////////////////////////

static const char*  sg_category_p = "jell";

#define INFO(fmt,arg...)    \
    do {   \
        char    _szBuf[1024];   \
        snprintf(_szBuf, sizeof(_szBuf) - 1, fmt, ##arg);   \
        this->logger->info(sg_category_p, _szBuf, __FILE__, __LINE__); \
    } while(0)

#define ERROR(fmt,arg...)    \
    do {   \
        char    _szBuf[1024];   \
        snprintf(_szBuf, sizeof(_szBuf) - 1, fmt, ##arg);   \
        this->logger->error(sg_category_p, _szBuf, __FILE__, __LINE__); \
    } while(0)

#define DEBUG(fmt,arg...)    \
    if (this->logger->debugable()) {   \
        char    _szBuf[1024];   \
        snprintf(_szBuf, sizeof(_szBuf) - 1, fmt, ##arg);   \
        this->logger->debug(sg_category_p, _szBuf, __FILE__, __LINE__); \
    }


///////////////////////////////////////////////////////////////////////
//
// Class Jelly Implementation
//


    ///////////////////////////////////////////////////////////////////

    Jelly::Jelly()
        :   jell_logger(new JellLogger()), logger(jell_logger),
            allocator(new JellAllocator()),
            context(new JellContext(this)),
            m_udper_p(NULL)
    {
    }

    Jelly::~Jelly()
    {
        // shutdown all socket listeners and connections
        this->shutdown();

        if (allocator) {
            delete allocator;
        }

        if (jell_logger) {
            delete jell_logger;
            jell_logger = NULL;
        }
    }

    ///////////////////////////////////////////////////////////////////

    ///////////////////////////////////////////////////////////////////

    void
    Jelly::connect(const Address& address) throw (JellException)
    {
        /*
         * STEPs :
         *  0. check if the address is a validate one
         *  1. check if the address already listened; if yes, just return
         *  2. create a new socket, and listen on it
         */
         address_type   addr_type = this->getAddressType(address);
         if (AT_UNKNOWN == addr_type) {
             THROW_FMT(ERROR::ADDRESS_PROTOCOL, "Invalid address '%s'.", address.uri().c_str());
         }

        std::string uri = address.uri();

        if (AT_TCP == addr_type) {
            table_tcp_transporter_type::iterator result = m_table_tcp_transporter.find( uri );
            if (m_table_tcp_transporter.end() != result) {
                // found
                return;
            }

            // NOT found, create a new listener
            try {
                TCPSocket*  socket_p = new TCPSocket( address.hostname(), address.port() );
                m_table_tcp_transporter[ uri ] = socket_p;
            }
            catch(std::exception& e) {
                THROW_FMT(ERROR::UDP_LISTEN, "%s", e.what());
            }
        }
        else if (AT_UDP == addr_type) {
            // No connection establed for UDP
        }
        else if (AT_UNIX == addr_type) {
            THROW(ERROR::NOT_IMPLEMENT, "Not Implement.");
        }
    }

    void
    Jelly::listen(const Address& address) throw (JellException)
    {
        /*
         * STEPs :
         *  0. check if the address is a validate one
         *  1. check if the address already listened; if yes, just return
         *  2. create a new socket, and listen on it
         */
         address_type   addr_type = this->getAddressType(address);
         if (AT_UNKNOWN == addr_type) {
             THROW_FMT(ERROR::ADDRESS_PROTOCOL, "Invalid address '%s'.", address.uri().c_str());
         }

        std::string uri = address.uri();

        if (AT_TCP == addr_type) {
            table_tcp_listener_type::iterator    result = m_table_tcp_listener.find( uri );
            if (m_table_tcp_listener.end() != result) {
                // found
                return;
            }

            // NOT found, create a new listener
            try {
                TCPServerSocket*  socket_p = new TCPServerSocket( address.port() );
                m_table_tcp_listener[ uri ] = socket_p;
                INFO("listen on TCP port %d.", address.port());
            }
            catch(std::exception& e) {
                THROW_FMT(ERROR::TCP_LISTEN, "%s", e.what());
            }
        }
        else if (AT_UDP == addr_type) {
            table_udp_listener_type::iterator    result = m_table_udp_listener.find( uri );
            if (m_table_udp_listener.end() != result) {
                // found
                return;
            }

            // NOT found, create a new listener
            try {
                UDPSocket*  socket_p = new UDPSocket( address.port() );
                m_table_udp_listener[ uri ] = socket_p;
                INFO("listen on UDP port %d.", address.port());
            }
            catch(std::exception& e) {
                THROW_FMT(ERROR::UDP_LISTEN, "%s", e.what());
            }
        }
        else if (AT_UNIX == addr_type) {
            THROW(ERROR::NOT_IMPLEMENT, "Not Implement.");
        }
    }

    void
    Jelly::shutdown(const Address& address) throw (JellException)
    {
         address_type   addr_type = this->getAddressType(address);
         if (AT_UNKNOWN == addr_type) {
             THROW_FMT(ERROR::ADDRESS_PROTOCOL, "Invalid address '%s'.", address.uri().c_str());
         }

        std::string uri = address.uri();
        if (AT_TCP == addr_type) {
            table_tcp_listener_type::iterator    result = m_table_tcp_listener.find( uri );
            if (m_table_tcp_listener.end() != result) {
                TCPServerSocket*  socket_p = result->second;
                delete socket_p;
                m_table_tcp_listener.erase( uri );
            }

            // TODO : to shutdown TCP transporter
        }
        else if (AT_UDP == addr_type) {
            table_udp_listener_type::iterator    result = m_table_udp_listener.find( uri );
            if (m_table_udp_listener.end() != result) {
                UDPSocket*  socket_p = result->second;
                delete socket_p;
                m_table_udp_listener.erase( uri );
            }
        }
        else if (AT_UNIX == addr_type) {
            table_tcp_listener_type::iterator    result = m_table_tcp_listener.find( uri );
            if (m_table_tcp_listener.end() != result) {
                TCPServerSocket*  socket_p = result->second;
                delete socket_p;
                m_table_tcp_listener.erase( uri );
            }

            // TODO : to shutdown TCP transporter
        }
    }

    void
    Jelly::shutdown() throw (JellException)
    {
        // shutdown UDP listener
        for (table_udp_listener_type::iterator it = m_table_udp_listener.begin();
             it != m_table_udp_listener.end();
             ++it ) {
            UDPSocket*  socket_p = it->second;
            delete socket_p;
        }
        m_table_udp_listener.clear();

        // shutdown TCP transporter
        for (table_tcp_transporter_type::iterator it = m_table_tcp_transporter.begin();
             it != m_table_tcp_transporter.end();
             ++it ) {
            TCPSocket*  socket_p = it->second;
            delete socket_p;
        }
        m_table_tcp_transporter.clear();

        // shutdown TCP listener
        for (table_tcp_listener_type::iterator it = m_table_tcp_listener.begin();
             it != m_table_tcp_listener.end();
             ++it ) {
            TCPServerSocket*  socket_p = it->second;
            delete socket_p;
        }
        m_table_tcp_listener.clear();
    }

    ///////////////////////////////////////////////////////////////////

    void
    Jelly::addBufHandler(BufHandler* handler) throw (JellException)
    {
        m_table_buf_handler.push_back( handler );
    }

    void
    Jelly::addEventHandler(EventHandler* handler) throw (JellException)
    {
        m_table_event_handler.push_back( handler );
    }

    ///////////////////////////////////////////////////////////////////

    void
    Jelly::run(void)  throw (JellException)
    {
        m_ev_signal_stop.set< Jelly, &Jelly::ev_signal_stop_cb >( this );
        m_ev_signal_stop.start( SIGINT );
        m_ev_signal_stop.start( SIGTERM );

        //m_ev_idle.set< Jelly, &Jelly::ev_idle_cb >( this );
        //m_ev_idle.start();    // maybe eat-up CPU

        m_ev_prepare.set< Jelly, &Jelly::ev_prepare_cb >( this );
        m_ev_prepare.start();

        // prepare UDP ioer
        {
            UDPSocket*  socket_p = NULL;
            for (table_udp_listener_type::iterator   it = m_table_udp_listener.begin();
                 it != m_table_udp_listener.end();
                 ++it )
            {
                const std::string&  uri = it->first;
                socket_p = it->second;
                this->prepareTransportWatcher( socket_p, uri );
            }
            if (NULL != m_udper_p) {
                socket_p = m_udper_p;
                this->prepareTransportWatcher( socket_p, UDP_ANY_URI );
            }
        }

        // prepare TCP Listener
        {
            TCPServerSocket*  socket_p = NULL;
            for (table_tcp_listener_type::iterator   it = m_table_tcp_listener.begin();
                 it != m_table_tcp_listener.end();
                 ++it )
            {
                const std::string&  uri = it->first;
                socket_p = it->second;
                this->prepareListenWatcher( socket_p, uri );
            }
        }

        // we are ready.
        // run !!!
        m_ev_loop = ev_default_loop(0);
        //ev_loop(m_ev_loop, 0);
        for(int i=0; i<20; i++) {
            fprintf(stderr, "%s.%d ev_loop #%d\n", __FILE__, __LINE__, i + 1);
            ev_loop(m_ev_loop, EVLOOP_ONESHOT);
        }
    }

    void
    Jelly::prepareTransportWatcher( UDPSocket* socket_p, const std::string& uri )
    {
        if (NULL == socket_p) {
            return;
        }

        int     fd = socket_p->descriptor();
        // todo : to check that this fd is NOT in lookup table
        table_udp_read_watcher_type::iterator  result = m_table_udp_read_watcher.find( uri );
        if (m_table_udp_read_watcher.end() != result) {
            return;
        }

        UDPWatcher* pReader = new UDPWatcher( socket_p );
        pReader->set< Jelly, &Jelly::ev_udp_read_cb >( this );
        pReader->start(fd, ev::READ);
        m_table_udp_read_watcher[ uri ] = pReader;

        UDPWatcher* pWriter = new UDPWatcher( socket_p );
        pWriter->set< Jelly, &Jelly::ev_udp_write_cb >( this );
        //pWriter->start(fd, ev::WRITE);
        m_table_udp_write_watcher[ uri ] = pWriter;
    }

    void
    Jelly::prepareListenWatcher( TCPServerSocket* socket_p, const std::string& uri )
    {
        if (NULL == socket_p) {
            return;
        }

        int     fd = socket_p->descriptor();
        // todo : to check that this fd is NOT in lookup table
        table_tcp_accept_watcher_type::iterator  result = m_table_tcp_accept_watcher.find( uri );
        if (m_table_tcp_accept_watcher.end() != result) {
            return;
        }

        TCPAcceptWatcher* pReader = new TCPAcceptWatcher( socket_p, uri );
        pReader->set< Jelly, &Jelly::ev_tcp_accept_cb >( this );
        pReader->start(fd, ev::READ);
        m_table_tcp_accept_watcher[ uri ] = pReader;
    }

    void
    Jelly::registerTransportWatcher( TCPSocket* socket_p, const std::string& uri )
    {
        if (NULL == socket_p) {
            return;
        }

        INFO("register transport watcher for %s", uri.c_str());
        int     fd = socket_p->descriptor();
        // todo : to check that this fd is NOT in lookup table
        table_tcp_read_watcher_type::iterator  result = m_table_tcp_read_watcher.find( uri );
        if (m_table_tcp_read_watcher.end() != result) {
            return;
        }

        TCPWatcher* pReader = new TCPWatcher( socket_p, uri );
        pReader->set< Jelly, &Jelly::ev_tcp_read_cb >( this );
        pReader->start(fd, ev::READ);
        m_table_tcp_read_watcher[ uri ] = pReader;

        TCPWatcher* pWriter = new TCPWatcher( socket_p, uri );
        pWriter->set< Jelly, &Jelly::ev_tcp_write_cb >( this );
        //pWriter->start(fd, ev::WRITE);
        m_table_tcp_write_watcher[ uri ] = pWriter;
    }

    void
    Jelly::unregisterTransportWatcher( const std::string& uri )
    {
        INFO("unregister transport watcher for %s", uri.c_str());
        {
            table_tcp_read_watcher_type::iterator  result = m_table_tcp_read_watcher.find( uri );
            if (m_table_tcp_read_watcher.end() != result) {
                TCPWatcher* reader_p = result->second;
                reader_p->stop();
                delete reader_p;
                m_table_tcp_read_watcher.erase(uri);
            }
        }
        {
            table_tcp_write_watcher_type::iterator  result = m_table_tcp_write_watcher.find( uri );
            if (m_table_tcp_write_watcher.end() != result) {
                TCPWatcher* writer_p = result->second;
                writer_p->stop();
                delete writer_p;
                m_table_tcp_write_watcher.erase(uri);
            }
        }
    }

    ///////////////////////////////////////////////////////////////////

    void
    Jelly::ev_tcp_read_cb(ev::io& watcher, int revents) throw()
    {
        TCPWatcher* watcher_p = static_cast< TCPWatcher* >(&watcher);
        TCPSocket*  socket_p = watcher_p->m_socket_p;

        watcher_p->stop();

        if (0 == watcher_p->m_stage_recv) {
            // prepare
            watcher_p->m_size_to_recv = PACKET_HEAD_SIZE;
            watcher_p->m_recv_p = watcher_p->m_head_recv;
            watcher_p->m_stage_recv = 1;
        }

        int size_recv = socket_p->recv(watcher_p->m_recv_p, watcher_p->m_size_to_recv);
        if (size_recv < 0) {
            ERROR("size-want=%d size-got=%d", watcher_p->m_size_to_recv, size_recv);
        }
        else if(0 == size_recv) {
            // disconnect
            uri_type    uri = watcher_p->m_uri;
            unregisterTransportWatcher( uri );
            return;
        }
        DEBUG("from: %s bytes-recv: %p %u '%s'", watcher_p->m_uri.c_str(), watcher_p->m_recv_p, size_recv, dump_hex(watcher_p->m_recv_p, size_recv).c_str());
        if (1 == watcher_p->m_stage_recv) {
            // read packet head
            watcher_p->m_recv_p = watcher_p->m_head_recv + size_recv;
            watcher_p->m_size_to_recv -= size_recv;
            if (watcher_p->m_size_to_recv < 1) {
                // extract sizeof packet body
                packet_verifyHead(watcher_p->m_head_recv);
                unsigned    body_size = packet_getBodySize(watcher_p->m_head_recv);
                DEBUG("from: %s PACKET HEAD %p DUMP: %s", watcher_p->m_uri.c_str(), watcher_p->m_head_recv, dump_hex(watcher_p->m_head_recv, PACKET_HEAD_SIZE).c_str());
                DEBUG("from: %s sizeof-body-to-read: %u", watcher_p->m_uri.c_str(), body_size);

                // allocate a buf for packet body
                watcher_p->m_buf_recv_p = this->allocator->allocate( body_size );
                watcher_p->m_buf_recv_p->size( body_size );

                watcher_p->m_recv_p = watcher_p->m_buf_recv_p->getWritePtr();
                watcher_p->m_size_to_recv = watcher_p->m_buf_recv_p->getSizeToWrite();

                // switch to stage 2
                watcher_p->m_stage_recv = 2;
            }
        }
        else if (2 == watcher_p->m_stage_recv) {
            // read packet body
            watcher_p->m_buf_recv_p->adjustWritePtr(size_recv);
            watcher_p->m_recv_p = watcher_p->m_buf_recv_p->getWritePtr();
            watcher_p->m_size_to_recv = watcher_p->m_buf_recv_p->getSizeToWrite();
            if (watcher_p->m_size_to_recv < 1) {
                // to push (Buf, from_address) into QueueIn
                Buf*    buf_p = watcher_p->m_buf_recv_p;
                message_type    msg = make_tuple( buf_p, watcher_p->m_uri );
                m_queue_to_handle.push( msg );
                watcher_p->m_buf_recv_p = NULL;
                DEBUG("from: %s data: '%s'", watcher_p->m_uri.c_str(), buf_p->data());

                // switch to stage 0
                watcher_p->m_stage_recv = 0;
            }
        }

        watcher_p->start(socket_p->descriptor(), ev::READ);
    }

    void
    Jelly::ev_tcp_write_cb(ev::io& watcher, int revents) throw()
    {
        TCPWatcher* watcher_p = static_cast< TCPWatcher* >( &watcher);
        TCPSocket*  socket_p = watcher_p->m_socket_p;

        watcher_p->stop();

        if (0 == watcher_p->m_stage_send) {
            // prepare
            if (watcher_p->m_queue_message_out.size() < 1) {
                // no more message to send out
                return;
            }

            watcher_p->m_buf_send_p = watcher_p->m_queue_message_out.front();
            packet_setBodySize(watcher_p->m_head_send, watcher_p->m_buf_send_p->size());
            watcher_p->m_send_p = watcher_p->m_head_send;
            watcher_p->m_size_to_send = PACKET_HEAD_SIZE;
            watcher_p->m_stage_send = 1;
        }

        int size_sent = socket_p->send(watcher_p->m_send_p, watcher_p->m_size_to_send);
        if (size_sent < 0) {
            ERROR("size-want-to-send=%d size-sent=%d", watcher_p->m_size_to_send, size_sent);
        }
        else if(0 == size_sent) {
            // disconnect
            uri_type    uri = watcher_p->m_uri;
            unregisterTransportWatcher( uri );
            return;
        }
        if (1 == watcher_p->m_stage_send) {
            // packet head
            watcher_p->m_send_p = watcher_p->m_head_send + size_sent;
            watcher_p->m_size_to_send -= size_sent;
            if (watcher_p->m_size_to_send < 1) {
                watcher_p->m_send_p = watcher_p->m_buf_send_p->getReadPtr();
                watcher_p->m_size_to_send = watcher_p->m_buf_send_p->getSizeToRead();

                // switch to stage 2
                watcher_p->m_stage_send = 2;
            }
        }
        else if (2 == watcher_p->m_stage_send) {
            // packet body
            watcher_p->m_buf_send_p->adjustReadPtr(size_sent);
            watcher_p->m_send_p = watcher_p->m_buf_send_p->getReadPtr();
            watcher_p->m_size_to_send = watcher_p->m_buf_send_p->getSizeToRead();
            if (watcher_p->m_size_to_send < 1) {
                watcher_p->m_queue_message_out.pop();
                Buf*    buf_p = watcher_p->m_buf_send_p;
                this->allocator->deallocate( buf_p );
                watcher_p->m_buf_send_p = NULL;

                // switch to stage 0
                watcher_p->m_stage_send = 0;
            }
        }

        watcher_p->start(socket_p->descriptor(), ev::WRITE);
    }

    void
    Jelly::ev_tcp_accept_cb(ev::io& watcher, int revents) throw()
    {
        TCPAcceptWatcher* pWatcher = static_cast< TCPAcceptWatcher* >( &watcher);
        TCPServerSocket*  server_p = pWatcher->m_socket_p;

        try {
            TCPSocket*  tcp_socket_p = server_p->accept();
            std::string     hostname_ = tcp_socket_p->getForeignAddress();
            unsigned short  port_ = tcp_socket_p->getForeignPort();
            char    szAddress[256];
            snprintf(szAddress, sizeof(szAddress) - 1, "tcp://%s:%d", hostname_.c_str(), port_);
            std::string uri( szAddress );

            m_table_tcp_transporter[ uri ] = tcp_socket_p;
            registerTransportWatcher(tcp_socket_p, uri);
            INFO("accept %s", szAddress);
        }
        catch(std::exception& e) {
            THROW_FMT(ERROR::TCP_ACCEPT, "%s", e.what());
        }
    }

    void
    Jelly::ev_udp_read_cb(ev::io& watcher, int revents) throw()
    {
        static  const unsigned  BUF_SIZE = 65536;

        UDPWatcher* pWatcher = static_cast< UDPWatcher* >( &watcher);

        int ret = 0;
        std::string peerHost;
        unsigned short  peerPort;

        Buf*    buf_p = this->allocator->allocate( BUF_SIZE );
        try {
            buf_p->rewindWrite();
            ret = pWatcher->m_socket_p->recvFrom(buf_p->getWritePtr(), buf_p->getSizeToWrite(), peerHost, peerPort);
        }
        catch(std::exception& e) {
            this->allocator->deallocate( buf_p );
            THROW_FMT(ERROR::UDP_RECV, "%s", e.what());
        }

        buf_p->size( ret );
        char    szAddress[128];
        snprintf(szAddress, sizeof(szAddress) - 1, "udp://%s:%u", peerHost.c_str(), peerPort);
        //JellAddress     address( szAddress );

        //fprintf(stderr, "%s:%d %s from: %s data: %s\n", __FILE__, __LINE__, __FUNC__, szAddress, buf_p->data());
        DEBUG("from: %s data: %s\n", szAddress, buf_p->data());
        // to push (Buf, from_address) into QueueIn
        uri_type        uri(szAddress);
        message_type    msg = make_tuple( buf_p, uri );
        m_queue_to_handle.push( msg );
    }

    void
    Jelly::ev_udp_write_cb(ev::io& watcher, int revents) throw()
    {
        UDPWatcher* pWatcher = static_cast< UDPWatcher* >( &watcher);
        UDPSocket*  socket_p = pWatcher->m_socket_p;

        while (pWatcher->m_queue_message_out.size() > 0) {
            Buf*        buf_p;
            uri_type    uri;
            tie(buf_p, uri) = pWatcher->m_queue_message_out.front();
            JellAddress address( uri );

            try {
                buf_p->rewindRead();
                socket_p->sendTo(buf_p->getReadPtr(), buf_p->getSizeToRead(), address.hostname(), address.port());
            }
            catch(std::exception& e) {
                THROW_FMT(ERROR::UDP_SEND, "%s", e.what());
            }

            pWatcher->m_queue_message_out.pop();
        }

        if (pWatcher->m_queue_message_out.size() < 1) {
            // send-out all message
            pWatcher->stop();
        }
    }

    void
    Jelly::ev_signal_stop_cb(ev::sig &watcher, int revents)
    {
        if ( (SIGINT  == watcher.signum) ||
             (SIGTERM == watcher.signum)
            ) {
            ev_unloop(m_ev_loop, EVUNLOOP_ALL);
        }
    }

    void
    Jelly::ev_idle_cb(ev::idle& watcher, int revents)
    {
    }

    void
    Jelly::ev_prepare_cb(ev::prepare& watcher, int revents)
    {
        //fprintf(stderr, "%s:%d %s revents=%d ---- THIS MESSAGE SHOULD NOT BE SEEN ----\n", __FILE__, __LINE__, __FUNC__, revents);
        //fprintf(stderr, "%s:%d %s length-of-queue-to-handle=%d\n", __FILE__, __LINE__, __FUNC__, m_queue_to_handle.size());
        DEBUG("revents=%d ---- THIS MESSAGE SHOULD NOT BE SEEN ----", revents);
        DEBUG("length-of-queue-to-handle=%d", m_queue_to_handle.size());

        this->handleIncomingBuf();
    }

    ///////////////////////////////////////////////////////////////////

    void
    Jelly::handleIncomingBuf( void ) throw ()
    {
        while (m_queue_to_handle.size() > 0) {
            Buf*        buf_p;
            uri_type    uri;
            tie(buf_p, uri) = m_queue_to_handle.front();

            for (table_buf_handler_type::iterator it = m_table_buf_handler.begin();
                 m_table_buf_handler.end() != it;
                 ++it ) {
                BufHandler* handler_p = *it;
                int ret = handler_p->handleBuf(*buf_p, uri, *context);
                if ( ret < 0 ) {
                    // ERROR
                }
                else if ( ret > 0 ) {
                    // STOP handling this buf
                    break;
                }
            }
            m_queue_to_handle.pop();
        }
    }

    ///////////////////////////////////////////////////////////////////

    void
    Jelly::send(const Address& address, Buf& buf, int timeout_in_seconds) throw (JellException)
    {
         address_type   addr_type = this->getAddressType(address);
         if (AT_UNKNOWN == addr_type) {
             THROW_FMT(ERROR::ADDRESS_PROTOCOL, "Invalid address '%s'.", address.uri().c_str());
         }

        if (AT_TCP == addr_type) {
            this->tcp_send(address.uri(), buf, timeout_in_seconds);
        }
        else if (AT_UDP == addr_type) {
            this->udp_send(address.uri(), buf, timeout_in_seconds);
        }
        else if (AT_UNIX == addr_type) {
        }
    }

    void
    Jelly::recv(Address& address, Buf& buf, int timeout_in_seconds) throw (JellException)
    {
    }

    void
    Jelly::tcp_send(const uri_type& uri, Buf& buf, int timeout_in_seconds) throw (JellException)
    {
        table_tcp_write_watcher_type::iterator  it_watcher = m_table_tcp_write_watcher.find(uri);
        if (m_table_tcp_write_watcher.end() == it_watcher) {
            THROW_FMT(ERROR::SOCKET, "No TCP Write Watcher for '%s'.", uri.c_str());
        }

        TCPWatcher* watcher_p = it_watcher->second;
        watcher_p->send( &buf );
        watcher_p->start( watcher_p->m_socket_p->descriptor(), ev::WRITE);
    }

    void
    Jelly::udp_send(const uri_type& uri, Buf& buf, int timeout_in_seconds) throw (JellException)
    {
        if (m_table_udp_write_watcher.size() < 1) {
            // ERROR : No UDP writer available
            THROW_FMT(ERROR::SOCKET, "No UDP Writer available.");
        }

        table_udp_write_watcher_type::iterator  it_watcher = m_table_udp_write_watcher.begin();
        UDPWatcher* pWriter = it_watcher->second;
        message_type    message( &buf, uri );
        pWriter->send( message );
        pWriter->start( pWriter->m_socket_p->descriptor(), ev::WRITE);
    }

    ////////////////////////////////////////////////////////////////////

    void
    Jelly::sendto(const Address& address, Buf& buf, int timeout_in_seconds) throw (JellException)
    {
         address_type   addr_type = this->getAddressType(address);
         if (AT_UNKNOWN == addr_type) {
             THROW_FMT(ERROR::ADDRESS_PROTOCOL, "Invalid address '%s'.", address.uri().c_str());
         }

        if (AT_TCP == addr_type) {
            this->tcp_sendto(address, buf, timeout_in_seconds);
        }
        else if (AT_UDP == addr_type) {
            this->udp_sendto(address, buf, timeout_in_seconds);
        }
        else if (AT_UNIX == addr_type) {
        }
    }

    void
    Jelly::recvfrom(Address& address, Buf& buf, int timeout_in_seconds) throw (JellException)
    {
        // todo : to detect which kind of communication should be used
        //        maybe, we need a select()
        if (m_table_tcp_transporter.size() > 0) {
            this->tcp_recvfrom(address, buf, timeout_in_seconds);
        }
        else {
            this->udp_recvfrom(address, buf, timeout_in_seconds);
        }

        /*
         address_type   addr_type = this->getAddressType(address);
         if (AT_UNKNOWN == addr_type) {
             THROW_FMT(ERROR::ADDRESS_PROTOCOL, "Invalid address '%s'.", address.uri().c_str());
         }

        if (AT_TCP == addr_type) {
        }
        else if (AT_UDP == addr_type) {
            this->udp_recvfrom(address, buf, timeout_in_seconds);
        }
        else if (AT_UNIX == addr_type) {
        }
        */
    }

    ///////////////////////////////////////////////////////////////////

    void
    Jelly::tcp_sendto(const Address& address, Buf& buf, int timeout_in_seconds) throw (JellException)
    {
        std::string uri = address.uri();

        TCPSocket*  socket_p = NULL;
        table_tcp_transporter_type::iterator result = m_table_tcp_transporter.find(uri);
        if (m_table_tcp_transporter.end() == result) {
            THROW_FMT(ERROR::TCP_SEND, "No TCP connection to '%s'.", uri.c_str());
        }
        socket_p = result->second;

        try {
            // send head
            unsigned char    packet_head[ PACKET_HEAD_SIZE ];
            unsigned    data_size = buf.size();
            packet_setBodySize(packet_head, data_size);
            DEBUG("PACKET HEAD DUMP '%s'", dump_hex(packet_head, sizeof(packet_head)).c_str());
            socket_p->sendn(packet_head, sizeof(packet_head), timeout_in_seconds);

            // send body
            buf.rewindRead();
            DEBUG("BUF size-to-send=%u", buf.getSizeToRead());
            while(buf.getSizeToRead() > 0) {
                void*       p = buf.getReadPtr();
                unsigned    size_to_send = buf.getSizeToRead();

                int ret = socket_p->send(p, size_to_send);
                if (ret < 0) {
                    THROW_FMT(ERROR::TCP_SEND, "TCP send failed.");
                }
                buf.adjustReadPtr( ret );
                DEBUG("PACKET DATA DUMP '%s'", dump_hex(p, ret).c_str());
            }
        }
        catch(std::exception& e) {
            THROW_FMT(ERROR::TCP_SEND, "%s", e.what());
        }
    }

    void
    Jelly::tcp_recvfrom(Address& address, Buf& buf, int timeout_in_seconds) throw (JellException)
    {
        if (m_table_tcp_transporter.size() < 1) {
            THROW(ERROR::SOCKET, "No TCP transporter available.");
        }

        TCPSocket*  socket_p = NULL;
        table_tcp_transporter_type::iterator result = m_table_tcp_transporter.begin();
        if (m_table_tcp_transporter.end() == result) {
            THROW(ERROR::SOCKET, "No TCP transporter available.");
        }
        const uri_type& uri = result->first;
        socket_p = result->second;

        unsigned char    packet_head[ PACKET_HEAD_SIZE ];
        try {
            // recv head
            socket_p->recvn(packet_head, sizeof(packet_head), timeout_in_seconds);
        }
        catch(std::exception& e) {
            THROW_FMT(ERROR::TCP_RECV, "%s", e.what());
        }
        DEBUG("PACKET HEAD DUMP '%s'", dump_hex(packet_head, sizeof(packet_head)).c_str());
        unsigned    data_size = packet_getBodySize(packet_head);

        //Buf*    buf_p = this->allocator->allocate(data_size);
        Buf*    buf_p = &buf;
        buf_p->size( data_size );
        buf_p->rewindWrite();
        try {
            // recv body
            socket_p->recvn(buf_p->getWritePtr(), buf_p->getSizeToWrite(), timeout_in_seconds);
            DEBUG("PACKET DATA DUMP '%s'", dump_hex(buf_p->getWritePtr(), buf_p->getSizeToWrite()).c_str());
        }
        catch(std::exception& e) {
            THROW_FMT(ERROR::TCP_RECV, "%s", e.what());
        }

        Address*    address_p = &address;
        JellAddress*    jell_address_p = static_cast< JellAddress * >( address_p );
        *jell_address_p = uri;
    }

    ///////////////////////////////////////////////////////////////////

    void
    Jelly::udp_sendto(const Address& address, Buf& buf, int timeout_in_seconds) throw (JellException)
    {
        std::string uri = address.uri();

        UDPSocket*  socket_p = NULL;
        if (m_table_udp_listener.size() > 0) {
            table_udp_listener_type::iterator   it = m_table_udp_listener.begin();
            socket_p = it->second;
        }
        else {
            if (NULL == m_udper_p) {
                m_udper_p = new UDPSocket();
            }
            socket_p = m_udper_p;
        }

        try {
            buf.rewindRead();
            socket_p->sendTo(buf.getReadPtr(), buf.getSizeToRead(), address.hostname(), address.port());
        }
        catch(std::exception& e) {
            THROW_FMT(ERROR::UDP_SEND, "%s", e.what());
        }
    }

    void
    Jelly::udp_recvfrom(Address& address, Buf& buf, int timeout_in_seconds) throw (JellException)
    {
        UDPSocket*  socket_p = NULL;
        if (m_table_udp_listener.size() > 0) {
            table_udp_listener_type::iterator   it = m_table_udp_listener.begin();
            socket_p = it->second;
        }
        else if (NULL != m_udper_p) {
            socket_p = m_udper_p;
        }
        else {
            THROW(ERROR::SOCKET, "No UDP listener available.");
        }

        int ret = 0;
        std::string peerHost;
        unsigned short  peerPort;

        try {
            buf.rewindWrite();
            ret = socket_p->recvFrom(buf.getWritePtr(), buf.getSizeToWrite(), peerHost, peerPort);
        }
        catch(std::exception& e) {
            THROW_FMT(ERROR::UDP_RECV, "%s", e.what());
        }

        buf.size( ret );
        char    szAddress[128];
        snprintf(szAddress, sizeof(szAddress) - 1, "udp://%s:%u", peerHost.c_str(), peerPort);
        Address*    address_p = &address;
        JellAddress*    jell_address_p = static_cast< JellAddress * >( address_p );
        *jell_address_p = szAddress;
    }

    ///////////////////////////////////////////////////////////////////

    Jelly::address_type
    Jelly::getAddressType(const Address& address) const
    {
        if (0 == strcmp(address.protocol(), TCP_NAME)) {
            return  AT_TCP;
        }
        if (0 == strcmp(address.protocol(), UDP_NAME)) {
            return  AT_UDP;
        }
        if (0 == strcmp(address.protocol(), UNIX_NAME)) {
            return  AT_UNIX;
        }

        return  AT_UNKNOWN;
    }

///////////////////////////////////////////////////////////////////////

static bool    packet_verifyHead(void* packet_head)
{
    char*   head_p = static_cast< char* >(packet_head);
    if (0 == memcmp(head_p, PACKET_BOM, PACKET_BOM_SIZE)) {
        return  true;
    }
    else {
        return  false;
    }
}

static unsigned  packet_getBodySize(void* packet_head)
{
    unsigned char*  head_p = static_cast< unsigned char* >(packet_head);
    unsigned char*  p = head_p + PACKET_BOM_SIZE;
    unsigned    body_size = ((((((p[0] << 8) + p[1]) << 8) + p[2]) << 8) + p[3]);

    return  body_size;
}

static void  packet_setBodySize(void* packet_head, unsigned data_size)
{
    char*   head = static_cast< char* >(packet_head);
    memmove(head, PACKET_BOM, PACKET_BOM_SIZE);
    head[PACKET_BOM_SIZE + 0] = (data_size & 0xFF000000) >> 24;
    head[PACKET_BOM_SIZE + 1] = (data_size & 0x00FF0000) >> 16;
    head[PACKET_BOM_SIZE + 2] = (data_size & 0x0000FF00) >> 8;
    head[PACKET_BOM_SIZE + 3] = (data_size & 0x000000FF);
}

static  std::string dump_hex(void* data, size_t size)
{
    unsigned char*  p = static_cast< unsigned char* >(data);
    std::string     result;
    const char*     pad_p = " ";

    while( size > 16 ) {
        char    hex_part[49];
        char*   h_p = hex_part;
        char    bin_part[17];
        char*   b_p = bin_part;

        for(int i=0; i<16; i++) {
            unsigned    ord = *(p + i);
            sprintf(h_p + i*3, "%02X ", ord);
            if ((ord >= 0x20) && (ord < 0x7F)) {
                b_p[i] = *(p + i);
            }
            else {
                b_p[i] = '.';
            }
        }
        bin_part[ sizeof(bin_part) - 1 ] = '\n';
        result.append(hex_part, sizeof(hex_part) - 1);
        result.append(pad_p, 1);
        result.append(bin_part, sizeof(bin_part));

        p += 16;
        size -= 16;
    }
    if (size > 0) {
        char    hex_part[49];
        char*   h_p = hex_part;
        char    bin_part[17];
        char*   b_p = bin_part;

        for(int i=0; i<16; i++) {
            if (i < size) {
                unsigned    ord = *(p + i);
                sprintf(h_p + i*3, "%02X ", ord);
                if ((ord >= 0x20) && (ord < 0x7F)) {
                    b_p[i] = *(p + i);
                }
                else {
                    b_p[i] = '.';
                }
            }
            else {
                sprintf(h_p + i*3, "   ");
                b_p[i] = ' ';
            }
        }
        bin_part[ sizeof(bin_part) - 1 ] = '\0';
        result.append(hex_part, sizeof(hex_part) - 1);
        result.append(pad_p, 1);
        result.append(bin_part, sizeof(bin_part));
    }

    return  result;
}


}   // namespace jell

